package com.ikas.autumn.iec104.module.iec.nettyconfig;


import cn.hutool.cache.impl.TimedCache;
import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.IdUtil;
import com.alibaba.fastjson.JSONObject;
import com.ikas.autumn.iec104.common.enums.AttrCategoryEnum;
import com.ikas.autumn.iec104.common.util.SpringContextHolder;
import com.ikas.autumn.iec104.common.util.date.DateTimeUtil;
import com.ikas.autumn.iec104.module.data.mysql.handler.ProducerHandler;
import com.ikas.autumn.iec104.module.iec.apdumodel.Apdu;
import com.ikas.autumn.iec104.module.iec.apdumodel.ApduCustom;
import com.ikas.autumn.iec104.module.iec.apdumodel.Asdu;
import com.ikas.autumn.iec104.module.iec.apdumodel.Cot;
import com.ikas.autumn.iec104.module.iec.apdunew.TransferReason;
import com.ikas.autumn.iec104.module.iec.asdunew.ie.InformationDetail;
import com.ikas.autumn.iec104.module.iec.asdunew.ie.InformationObject;
import com.ikas.autumn.iec104.module.iec.asdunew.internal.CauseOfTransmission;
import com.ikas.autumn.iec104.module.iec.container.Iec104Link;
import com.ikas.autumn.iec104.module.iec.container.LinkContainer;
import com.ikas.autumn.iec104.module.iec.netty.AbstractTcpMasterBuilder;
import com.ikas.autumn.iec104.module.iec.utils.ByteUtil;
import com.ikas.autumn.iec104.module.iec.utils.DataConvertor;
import com.ikas.autumn.iec104.module.protocol.entity.AnswerTsCode;
import com.ikas.autumn.iec104.module.protocol.entity.ApduReceiveData;
import com.ikas.autumn.iec104.module.protocol.entity.Connection;
import com.ikas.autumn.iec104.module.protocol.enums.ProtocolEnum;
import com.ikas.autumn.iec104.module.protocol.handler.ProtocolHandler;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * 消息处理类
 *
 * @author 修唯xiuwei
 * @version 3.0
 */
@Slf4j
@NoArgsConstructor
public class Master104Handle extends SimpleChannelInboundHandler<ByteBuf> {

    /**
     * 遥信数据的缓存key
     */
    private static final String BASE_METE_UNIQUE_YX_IDS = "mete:base:unique:yx:ids";
    /**
     * 遥测数据的缓存key
     */
    private static final String BASE_METE_UNIQUE_YC_IDS = "mete:base:unique:yc:ids";

    private ProtocolHandler protocolHandler;

    /**
     * Master 104 handle
     *
     * @param masterBuilder master builder
     */
    public Master104Handle(AbstractTcpMasterBuilder masterBuilder) {
        this.masterBuilder = masterBuilder;
    }

    public Master104Handle(AbstractTcpMasterBuilder masterBuilder, ProtocolHandler protocolHandler) {
        this.masterBuilder = masterBuilder;
        this.protocolHandler = protocolHandler;
    }

    public Master104Handle(AbstractTcpMasterBuilder masterBuilder, Class<? extends Apdu> apduClass) {
        this(masterBuilder);
        this.apduClass = apduClass;
    }

    private AbstractTcpMasterBuilder masterBuilder;

    private int testNum;

    private int exceptionNum;

    /**
     * 是否是主动断开连接
     */
    private boolean isInitiative;

    private Class<? extends Apdu> apduClass = Apdu.class;

    /**
     * 通信管道读取信息
     *
     * @param ctx the context
     * @param msg the message
     * @throws Exception
     */
    @Override
    public void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        //打印接收的报文
        String receiveCode = logInfoReceiveMsg(msg);
        log.info("1.1开始解析报文");
        //收数据
        Apdu apdu = apduClass.newInstance()
                .setChannel(ctx.channel())
                .setIec104Builder(masterBuilder)
                .setLog(log)
                .setReceiveCode(receiveCode)
                .loadByteBuf(msg);
        if (apdu.getApciType() == Apdu.ApciType.I_FORMAT) {
            this.testNum = 0;
        }
        //回复信息
        apdu.answer(apdu);
        //处理接收到的数据
        List<ApduCustom> customs = getApduCustom(apdu);
        if (!CollectionUtils.isEmpty(customs)) {
            log.info("1.2.1接收的遥信遥测值结果:{}", JSONObject.toJSONString(customs));
//            List<ApduCustom> finalCustoms = customs;
            //异步执行保存数据
//            CompletableFuture.runAsync(() -> {
//                try {
//                    //是否保存响应总召唤报文
//                    saveAnswerTsCode(apdu);
//                    //是否保存解析数据
//                    saveApduCustoms(finalCustoms);
//                } catch (Exception e) {
//                    log.error("保存响应总召唤报文异常{}", e.getMessage());
//                }
//            });
//            //过滤信息
            customs = filterCustomers(customs);
            //发送kafka
            yxYcSendKafka(customs);
        } else {
            log.info("1.2.2没有解析出有效的遥信遥测值信息");
        }
    }

    /**
     * 是否保存响应总召唤报文
     *
     * @param apdu 解析报文后的对象
     */
    private void saveAnswerTsCode(Apdu apdu) {
        //保存总召唤报文开关
        Boolean saveTsCodeFlag = SpringContextHolder.getBean(ProducerHandler.class)
                .getSaveTsCodeFlag();
        String receiveCode = apdu.getReceiveCode();
        Asdu asdu = apdu.getAsdu();
        if (Boolean.TRUE.equals(saveTsCodeFlag) &&
                asdu != null) {
            //传送原因 过滤出总召唤的报文
            Cot cot = asdu.getCot();
            if (cot != null) {
                CauseOfTransmission transmission = CauseOfTransmission.causeFor(cot.getNot());
                AnswerTsCode answerTs = new AnswerTsCode();
                answerTs.setTsCode(receiveCode);
                answerTs.setTsType(asdu.getTypeId());
                answerTs.setTsTypeDesc(asdu.getASduType().getDescription() + "(" + transmission.getDesc() + ")");
//                Long batchNum = RedisUtil.getLong(TechnicalTerm.REDIS_TOTAL_SUMMON_KEY);
//                answerTs.setBatchNum(batchNum);
//                SpringContextHolder.getBean(AnswerTsCodeDAO.class).insert(answerTs);
                log.info("保存总召唤报文信息：{}", JSONObject.toJSONString(answerTs));
            }
        }

    }

    /**
     * 保存解析数据
     *
     * @param customs 接收的数据集
     */
    private void saveApduCustoms(List<ApduCustom> customs) {
        if (CollectionUtils.isEmpty(customs)) {
            return;
        }
        Boolean saveTsDataFlag = SpringContextHolder.getBean(ProducerHandler.class)
                .getSaveTsDataFlag();
        Boolean saveReceiveDataFlag = SpringContextHolder.getBean(ProducerHandler.class)
                .getSaveReceiveDataFlag();
        Boolean saveYxReceiveDataFlag = SpringContextHolder.getBean(ProducerHandler.class)
                .getSaveYxReceiveDataFlag();
        Boolean saveYcReceiveDataFlag = SpringContextHolder.getBean(ProducerHandler.class)
                .getSaveYcReceiveDataFlag();

        List<ApduReceiveData> reDatas = customs.stream()
                .map(this::coverToReceiveData).collect(Collectors.toList());
        //不保留总召唤数据
        if (Boolean.FALSE.equals(saveTsDataFlag)) {
            log.info("准备保存数据时，过滤掉接收的总召唤数据");
            reDatas = reDatas.stream()
                    .filter(r -> r.getCotType() != TransferReason.ANSWER_STATION_CALL.getCode()).collect(Collectors.toList());
        }
        //不保留除总召唤外的接收数据
        if (Boolean.FALSE.equals(saveReceiveDataFlag)) {
            log.info("准备保存数据时，过滤掉接收的除了总召唤外的数据");
            reDatas = reDatas.stream()
                    .filter(r -> r.getCotType() == TransferReason.ANSWER_STATION_CALL.getCode()).collect(Collectors.toList());
        }
        if (Boolean.FALSE.equals(saveYxReceiveDataFlag)) {
            log.info("准备保存数据时，过滤掉接收的遥信数据");
            reDatas = reDatas.stream()
                    .filter(r -> r.getCategoryType() != AttrCategoryEnum.YX.getCode())
                    .collect(Collectors.toList());
        }
        if (Boolean.FALSE.equals(saveYcReceiveDataFlag)) {
            log.info("准备保存数据时，过滤掉接收的遥测数据");
            reDatas = reDatas.stream()
                    .filter(r -> r.getCategoryType() != AttrCategoryEnum.YC.getCode())
                    .collect(Collectors.toList());
        }

        if (!CollectionUtils.isEmpty(reDatas)) {
            log.info("保存receive_data接收到的数据：{}", JSONObject.toJSONString(reDatas));
//            SpringContextHolder.getBean(ApduReceiveDataServiceImpl.class)
//                    .saveBatch(reDatas);
        }

    }

    private ApduReceiveData coverToReceiveData(ApduCustom custom) {
        ApduReceiveData apduReData = BeanUtil.copyProperties(custom, ApduReceiveData.class);
        if (custom.getGenerateTime() != null) {
            String dateTimeString =
                    DateTimeUtil.getDateTimeString(
                            new Date(custom.getGenerateTime()), true
                    );
            apduReData.setDataTime(dateTimeString);
        }
//        if (TransferReason.ANSWER_STATION_CALL.getCode() == custom.getCotType()) {
//            apduReData.setIsTotalSummon(1);
//            Long batchNum = RedisUtil.getLong(TechnicalTerm.REDIS_TOTAL_SUMMON_KEY);
//            apduReData.setTsBatchNum(batchNum);
//        }
        return apduReData;
    }

    /**
     * 过滤信息
     *
     * @param customs 解析的列表
     */
    private List<ApduCustom> filterCustomers(List<ApduCustom> customs) {
        Boolean filterTsFlag = SpringContextHolder.getBean(ProducerHandler.class)
                .getFilterTsFlag();
        //过滤掉总召唤返回的数据
        if (Boolean.TRUE.equals(filterTsFlag)) {
            log.info("2.1.1准备发送kafka时，开始过滤总召唤数据");
            customs = customs.stream()
                    .filter(c -> c.getCotType() != TransferReason.ANSWER_STATION_CALL.getCode())
                    .collect(Collectors.toList());
        } else {
            //总召唤 遥测值无时标时添加系统时间
            log.info("2.1.2筛选总召唤中遥测值无时标时添加系统时间");
            customs = customs.stream().peek(c -> {
                if (TransferReason.ANSWER_STATION_CALL.getCode() == c.getCotType()
                        && AttrCategoryEnum.YC.getCode() == c.getCategoryType()
                        && c.getGenerateTime() == null) {
                    c.setGenerateTime(System.currentTimeMillis());
                    c.setIsSystemTime(1);
                }
            }).collect(Collectors.toList());
        }
        List<ApduCustom> notDateTimeDs = customs.stream()
                //打印无时标数据
                .filter(c -> c.getGenerateTime() == null).collect(Collectors.toList());
        if (!CollectionUtils.isEmpty(notDateTimeDs)) {
            log.error("接收到的无时标数据：{}", notDateTimeDs);
        }
        //过滤掉没有时间格式的数据
        log.info("2.2准备发送kafka时，开始过滤掉没有时间格式的数据");

        //开始过滤和业务表相关的测点数据
//        sendList = filterLinkOfBusiness(sendList);
        return customs.stream()
                //过滤带时标数据
                .filter(c -> c.getGenerateTime() != null)
                .collect(Collectors.toList());
    }

    /**
     * 过滤和业务相关的数据
     *
     * @param sendList 准备发送的数据
     */
    private List<ApduCustom> filterLinkOfBusiness(List<ApduCustom> sendList) {
        TimedCache<String, List<Long>> yxLocalCache = SpringContextHolder.getBean(ProducerHandler.class).getYxLocalCache();
        TimedCache<String, List<Long>> ycLocalCache = SpringContextHolder.getBean(ProducerHandler.class).getYcLocalCache();
        //检查和初始化遥信遥测本地缓存
        checkAndInitYxYcLocalCache(yxLocalCache, ycLocalCache);
        log.info("3.1开始过滤业务相关数据，原数据size:{}", sendList.size());
        long ycCount = sendList.stream().filter(p -> p.getCategoryType() == AttrCategoryEnum.YC.getCode()).count();
        long yxCount = sendList.stream().filter(p -> p.getCategoryType() == AttrCategoryEnum.YX.getCode()).count();
        List<Long> baseMeteYx = new ArrayList<>();
        List<Long> baseMeteYc = new ArrayList<>();
        if (ycCount > 0) {
            baseMeteYc = ycLocalCache.get(BASE_METE_UNIQUE_YC_IDS);
            log.info("3.2.1获取本地缓存中业务遥测数据size{}", baseMeteYc == null ? 0 : baseMeteYc.size());
        }
        if (yxCount > 0) {
            baseMeteYx = yxLocalCache.get(BASE_METE_UNIQUE_YX_IDS);
            log.info("3.2.2获取本地缓存中业务遥信数据size{}", baseMeteYx == null ? 0 : baseMeteYx.size());
        }
        if (CollectionUtils.isEmpty(baseMeteYx) && CollectionUtils.isEmpty(baseMeteYc)) {
            log.info("3.2.3从本地缓存未获取到有效的遥信遥测数据, 接收数据不发kafka");
            return new ArrayList<>();
        }
        List<Long> finalBaseMeteYc = baseMeteYc;
        List<Long> finalBaseMeteYx = baseMeteYx;
        List<ApduCustom> filterList = sendList.stream().filter(
                s -> (s.getCategoryType() == AttrCategoryEnum.YC.getCode() && Objects.requireNonNull(finalBaseMeteYc).contains((long) s.getAddress())) ||
                        (s.getCategoryType() == AttrCategoryEnum.YX.getCode() && Objects.requireNonNull(finalBaseMeteYx).contains((long) s.getAddress()))

        ).collect(Collectors.toList());
        log.info("3.3过滤业务相关数据完成，过滤后数据size:{}", filterList.size());
        return filterList;
    }

    /**
     * 检查和初始化遥信遥测本地缓存
     */
    private void checkAndInitYxYcLocalCache(TimedCache<String, List<Long>> yxLocalCache, TimedCache<String, List<Long>> ycLocalCache) {
        List<Long> yxList = yxLocalCache.get(BASE_METE_UNIQUE_YX_IDS);
        List<Long> ycList = ycLocalCache.get(BASE_METE_UNIQUE_YC_IDS);
        if (CollectionUtils.isEmpty(yxList) ) {
            initYxLocalCache(yxLocalCache);
        }
        if (CollectionUtils.isEmpty(ycList) ) {
            initYcLocalCache(ycLocalCache);
        }
    }

    /**
     * 初始化遥信内部缓存
     */
    private void initYxLocalCache(TimedCache<String, List<Long>> yxLocalCache) {
//        List<Long> baseMeteYx = ObjectUtils.defaultIfNull(RedisUtil.get(BASE_METE_UNIQUE_YX_IDS), new ArrayList<>());
//        log.info("3.0初始化遥信本地缓存，来源redis，数据size：{}", baseMeteYx.size());
//        yxLocalCache.put(BASE_METE_UNIQUE_YX_IDS, baseMeteYx);
    }
    /**
     * 初始化遥测内部缓存
     */
    private void initYcLocalCache(TimedCache<String, List<Long>> ycLocalCache) {
//        List<Long> baseMeteYc = ObjectUtils.defaultIfNull(RedisUtil.get(BASE_METE_UNIQUE_YC_IDS), new ArrayList<>());
//        log.info("3.0初始化遥测本地缓存，来源redis，数据size：{}", baseMeteYc.size());
//        ycLocalCache.put(BASE_METE_UNIQUE_YC_IDS, baseMeteYc);
    }
    /**
     * 打印接收报文信息
     *
     * @param msg 消息通道的信息
     */
    private String logInfoReceiveMsg(ByteBuf msg) {
        ByteBuf copyByte = msg.duplicate();
        int i = copyByte.readableBytes();
        byte[] bytes = new byte[i];
        copyByte.readBytes(bytes);
        String receiveCode = ByteUtil.byteArray2HexString(bytes, bytes.length, true);
        log.info("收到报文: {}", receiveCode);
        return receiveCode;
    }

    /**
     * 从apdu中获取需要的列表对象
     *
     * @param apdu 解析报文后的对象
     * @return list ofr apduCustom
     */
    public static List<ApduCustom> getApduCustom(Apdu apdu) {
        ApduCustom comCustom = new ApduCustom();
        comCustom.setAcceptSeqNum(apdu.getReceiveSeqNum());
        comCustom.setSendSeqNum(apdu.getSendSeqNum());
        comCustom.setApciType(apdu.getApciType().name());
        Asdu aSdu = apdu.getAsdu();
        if (aSdu == null) {
            return new ArrayList<>();
        }
        comCustom.setAsduTypeId(aSdu.getASduType().getId());
        comCustom.setAsduTypeDesc(aSdu.getASduType().getDescription());
        int categoryType = aSdu.getASduType().getCategoryType();
        comCustom.setCategoryType(categoryType);
        Cot cot = aSdu.getCot();
        if (cot != null) {
            CauseOfTransmission causeOfTransmission = CauseOfTransmission.causeFor(cot.getNot());
            comCustom.setCotType(causeOfTransmission.getId());
            comCustom.setCotTypeDesc(causeOfTransmission.getDesc());
        }

        List<ApduCustom> customs = new ArrayList<>();
        InformationObject[] infoObjs = aSdu.getInformationObjects();
        if (categoryType != 1 && infoObjs != null) {
            for (InformationObject infoObj : infoObjs) {
                int ioa = infoObj.getInformationObjectAddress();
                List<InformationDetail> infoDetails = infoObj.getInformationDetail();
                if (!CollectionUtils.isEmpty(infoDetails)) {
                    infoDetails.forEach(detail -> {
                        ApduCustom custom = BeanUtil.copyProperties(comCustom, ApduCustom.class);
                        custom.setCategoryValue(detail.getValue());
                        custom.setAddress(ioa);
                        Long dateTime = detail.getDateTime();
                        custom.setGenerateTime(dateTime);
                        if (dateTime != null) {
                            custom.setGenerateDate(DateUtil.format(DateUtil.date(detail.getDateTime()), DatePattern.NORM_DATETIME_MS_PATTERN));
                        }
                        customs.add(custom);
                    });
                }
            }

        }
        return customs;
    }

    /**
     * 遥信遥测发送kafka
     *
     * @param customs 自定义的apdu
     */
    private void yxYcSendKafka(List<ApduCustom> customs) {
        if (CollectionUtils.isEmpty(customs)) {
            log.info("4.1.1-mt-104 没有找到有时标的可以发送的遥测或遥信数据");
        } else {
            String customsJson = JSONObject.toJSONString(customs);
            String key = IdUtil.simpleUUID();
            log.info("4.1.2-mt-104【准备往kafka发送的数据】key:{}, value:{}", key, customsJson);
            SpringContextHolder.getBean(ProducerHandler.class)
                    .sendYxYcKafka(key, customsJson);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        if (this.exceptionNum > 10) {
            this.isInitiative = true;
            reconnect(ctx);
        }
        this.exceptionNum++;
        ctx.flush();
        log.error("发生{}次异常，异常内容：", this.exceptionNum, cause);

    }


    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        this.isInitiative = false;
        InetSocketAddress ipSocket = (InetSocketAddress) ctx.channel().remoteAddress();
        InetSocketAddress localIpSocket = (InetSocketAddress) ctx.channel().localAddress();
        String clientIp = ipSocket.getAddress().getHostAddress();
        Integer clientPort = ipSocket.getPort();
        log.info("连接" + clientIp + ":" + clientPort + "服务端成功，本地端口：" + localIpSocket.getPort());
        LinkContainer.getInstance().getLinks().put(ctx.channel().id(), new Iec104Link(ctx.channel(), clientIp, clientPort, Iec104Link.Role.SLAVER));
        ctx.writeAndFlush(Unpooled.copiedBuffer(TechnicalTerm.START));

        this.masterBuilder.connected();
        protocolHandler.active(new Connection(ProtocolEnum.IEC, this.masterBuilder.getIp(), this.masterBuilder.getPort().toString()));
    }


    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        log.debug("第{}次发送测试帧：" + DataConvertor.byte2String(TechnicalTerm.TEST), testNum);
        ctx.writeAndFlush(Unpooled.copiedBuffer(TechnicalTerm.TEST));
        IdleStateEvent i = (IdleStateEvent) evt;
        if (!i.isFirst()) {
            log.debug("链路长时间无响应，重新连接");
            this.isInitiative = true;
            reconnect(ctx);
        }
        this.testNum++;
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        if (isInitiative) {
            this.isInitiative = false;
        } else {
            reconnect(ctx);
        }
        this.masterBuilder.disconnected();
        protocolHandler.inActive(new Connection(ProtocolEnum.IEC, this.masterBuilder.getIp(), this.masterBuilder.getPort().toString()));


    }


    private synchronized void reconnect(ChannelHandlerContext ctx) throws Exception {
        LinkContainer.getInstance().getLinks().remove(ctx.channel().id());
        ctx.close();
        if (masterBuilder != null) {
            this.testNum = 0;
            this.exceptionNum = 0;
            log.error(masterBuilder.getIp() + ":" + masterBuilder.getPort() + "断线,尝试重连");
            masterBuilder.getOrCreateConnectionListener().operationComplete(masterBuilder.getFuture());
        }
    }
}
